package com.dr.leo.sparkspring.service

import com.alibaba.fastjson.JSONArray
import com.dr.leo.sparkspring.util.ObjGenerator
import org.apache.spark.SparkContext
import org.apache.spark.sql.SparkSession
import org.springframework.beans.factory.annotation.Autowired
import org.springframework.stereotype.Service

/**
  * @author leo.jie (weixiao.me@aliyun.com)
  * @organization DataReal
  * @version 1.0
  * @website https://www.jlpyyf.com
  * @date 2019-08-05 21:42
  * @since 1.0
  */
@Service
class WordCountService {
  @Autowired
  private val sparkSession: SparkSession = null

  def wordCount(filePath: String): JSONArray = {
    val sc: SparkContext = sparkSession.sparkContext
    val fileRdd = sc.textFile(filePath).flatMap(_.split(" "))
      .map((_, 1))
      .reduceByKey(_ + _)
    val wordCountArray: JSONArray = new JSONArray()
    fileRdd.collect().foreach(x => {
      wordCountArray.add(ObjGenerator.newJSON(Seq((x._1, x._2)): _*))
    })
    wordCountArray
  }
}
